package com.shujia.flink.source;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.text.SimpleDateFormat;
import java.util.Date;

public class Demo05MySQLSource {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<Student> mysqlDS = env.addSource(new MySQLSource(10));
        mysqlDS.print();

        env.execute();
    }
}

class Student {
    String id;
    String name;
    Integer age;
    String gender;
    String clazz;

    public Student(String id, String name, Integer age, String gender, String clazz) {
        this.id = id;
        this.name = name;
        this.age = age;
        this.gender = gender;
        this.clazz = clazz;
    }

    @Override
    public String toString() {
        return "Student{" +
                "id='" + id + '\'' +
                ", name='" + name + '\'' +
                ", age=" + age +
                ", gender='" + gender + '\'' +
                ", clazz='" + clazz + '\'' +
                '}';
    }
}

class MySQLSource implements SourceFunction<Student> {

    Integer sleepTime;

    public MySQLSource(int sleepTime) {
        this.sleepTime = sleepTime;
    }


    @Override
    public void run(SourceContext<Student> ctx) throws Exception {

        Connection conn = DriverManager.getConnection("jdbc:mysql://master:3306/studentdb", "root", "123456");

        PreparedStatement pSt = conn.prepareStatement("select id,name,age,gender,clazz from student where last_mod >= ?");
        SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

        while (true) {
            String timeStr = df.format(new Date());
            System.out.println("当前获取数据的时间为：" + timeStr);

            pSt.setString(1, timeStr);
            ResultSet rs = pSt.executeQuery();

            while (rs.next()) {
                String id = rs.getString("id");
                String name = rs.getString("name");
                int age = rs.getInt("age");
                String gender = rs.getString("gender");
                String clazz = rs.getString("clazz");
                Student student = new Student(id, name, age, gender, clazz);

                ctx.collect(student);
            }

            Thread.sleep(sleepTime * 1000);
        }


    }

    @Override
    public void cancel() {

    }
}